package com.dyh.quickdemo.flux;

import com.alibaba.fastjson.JSON;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.GroupedFlux;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
public class TestFlux {
    @Test
    public void test(){
        log.info(Thread.currentThread().getName());
        Flux.defer(()->Flux.just(1,2,3))
                .doOnSubscribe(subscription ->  log.info("线程:{} {}",Thread.currentThread().getName(),subscription))
                .subscribe(e-> log.info("线程:{} {}",Thread.currentThread().getName(),e));
    }

    @Test
    public void test2(){
        log.info(Thread.currentThread().getName());
        Flux.just(1,2,3)
                .doOnSubscribe(subscription ->  log.info("线程:{} {}",Thread.currentThread().getName(),subscription))
                .subscribe(e-> log.info("线程:{} {}",Thread.currentThread().getName(),e));
    }

    @Test
    public void test3(){
        /*两个线程都订阅 相当于广播*/
        log.info(Thread.currentThread().getName());
        Flux source = Flux.defer(()->Flux.just(1,2,3,4))
                .doOnSubscribe(subscription ->  log.info("线程:{} {}",Thread.currentThread().getName(),subscription));

        Runnable r = ()->source.subscribe(e-> log.info("线程:{} {}",Thread.currentThread().getName(),e));
        new Thread(r).start();
        new Thread(r).start();

    }

    @Test
    public void test4(){

        log.info(Thread.currentThread().getName());

        Flux<GroupedFlux<Integer,Integer>> source = Flux.defer(()->Flux.just(1,2,3,4))
                .groupBy(e->e%2)
                .doOnSubscribe(subscription ->  log.info("线程:{} {}",Thread.currentThread().getName(),subscription));


       // source.subscribe(e ->  log.info("线程:{} {}",Thread.currentThread().getName(),e));//UnicastGroupedFlux


//        Thread[] threads=new Thread[]{ new Thread(),new Thread()};
        ExecutorService[] pools = {Executors.newFixedThreadPool(1),Executors.newFixedThreadPool(1)};


       // source.subscribe(f -> f.subscribe(t->log.info("{} {} {}",Thread.currentThread().getName(),f.key(),t)));//main
        source.subscribe(f -> f.subscribe(t->{
            log.info("{} 添加订阅 {}",Thread.currentThread().getName(),t);
            pools[f.key()].submit(()->log.info("线程:{} 参数:{}",Thread.currentThread().getName(),t));
        }));//main
//        Runnable r = ()->source.subscribe(e-> log.info("线程:{} {}",Thread.currentThread().getName(),e));
//        new Thread(r).start();
//        new Thread(r).start();

    }

    /*简单demo 分组后多个线程消费各种组的*/
    @Test
    public void test5(){
        log.info(Thread.currentThread().getName());
        Flux<GroupedFlux<Integer,Integer>> source = Flux.defer(()->Flux.just(1,2,3,4))
                .groupBy(e->e%2)
                .doOnSubscribe(subscription ->  log.info("线程:{} {}",Thread.currentThread().getName(),subscription));
        ExecutorService[] pools = {Executors.newFixedThreadPool(1),Executors.newFixedThreadPool(1)};

        source.subscribe(f -> f.subscribe(t->{
            log.info("{} 添加订阅 {}",Thread.currentThread().getName(),t);//所有都要的操作可以加在这里
            pools[f.key()].submit(()->log.info("线程:{} 参数:{}",Thread.currentThread().getName(),t));
        }));
    }

    @Test
    public void test6(){
        StringBuilder sb1 = new StringBuilder("1");
        StringBuilder sb2 = new StringBuilder("2");

        System.out.println(sb1.append(sb2));
    }

    @Test
    public void test7(){
        String js = "{\"a\":1,\"c\":2}";
        Js j = JSON.parseObject(js,Js.class);
        /*ok*/
        System.out.println();
    }

    @Setter@Getter
    static class Js{
        String a;
        String c;
    }

    class MyThread extends Thread{

    }
}
